package spark.Task2;


import org.apache.spark.sql.SparkSession;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;


public class Task2{
    public static void main(String[] args){
        SparkConf conf = new SparkConf().setAppName("Task2").setMaster("local");
        try (JavaSparkContext task2 = new JavaSparkContext(conf)) {
            SparkSession spark = SparkSession
                    .builder()
                    .appName("task2")
                    .master("local")
                    .getOrCreate();

            Dataset<Row> stock_small = spark.read().format("csv").option("header", "true").load("file:///home/jinhaitao/stock.csv");
            Dataset<Row> dividends = spark.read().format("csv").option("header", "true").load("file:///home/jinhaitao/dividends_small.csv");
            
            stock_small.createOrReplaceTempView("stock");
            dividends.createOrReplaceTempView("dividends");

            // 第一小问
            Dataset<Row> a = spark.sql("SELECT dividends.date, dividends.symbol, stock.stock_price_close FROM stock RIGHT JOIN dividends ON dividends.symbol = stock.stock_symbol AND dividends.date = stock.date ");  
            //a.show();
            a.createOrReplaceTempView("a");

            Dataset<Row> q1 = spark.sql("SELECT * FROM a WHERE symbol = 'IBM'");
            q1.show();
            q1.javaRDD().coalesce(1).saveAsTextFile("/home/jinhaitao/result/Task2_1");
 
            // 第二小问
            Dataset<Row> b = spark.sql("SELECT stock_price_adj_close, substring(date,1,4) as year FROM stock WHERE stock_symbol = 'AAPL'");
            //b.show();
            b.createOrReplaceTempView("b");
            Dataset<Row> b1 = spark.sql("SELECT year,AVG(stock_price_adj_close) as avg FROM b GROUP BY year HAVING  avg >=50 ORDER BY year");
            //b1.show();
            b1.javaRDD().coalesce(1).saveAsTextFile("/home/jinhaitao/result/Task2_2");

            spark.stop();
            task2.stop();
        }
    }

}
